package com.xnyzc.lhy.order.mq;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.xnyzc.lhy.common.component.config.security.RocketConfig;
import com.xnyzc.lhy.common.component.redis.RedisCommon;
import com.xnyzc.lhy.common.constant.MqConstant;
import com.xnyzc.lhy.common.constant.MqConsumerGroupConstant;
import com.xnyzc.lhy.common.constant.RedisPrefixConstant;
import com.xnyzc.lhy.common.constant.TimeLimitConstant;
import com.xnyzc.lhy.common.entity.Qw;
import com.xnyzc.lhy.common.entity.netty.NettyDriverOrderInfo;
import com.xnyzc.lhy.common.entity.netty.StriveCallbackParam;
import com.xnyzc.lhy.common.exception.DiDiException;
import com.xnyzc.lhy.common.penum.EDiDiErrorCode;
import com.xnyzc.lhy.common.penum.EDispatchType;
import com.xnyzc.lhy.common.penum.order.EOrderStatus;
import com.xnyzc.lhy.common.util.CheckUtil;
import com.xnyzc.lhy.common.util.IDUtil;
import com.xnyzc.lhy.nio.entity.netty.NettyUserOrderInfo;
import com.xnyzc.lhy.order.entity.common.order.OrderCommon;
import com.xnyzc.lhy.order.entity.gps.OaDVehicleGps;
import com.xnyzc.lhy.order.entity.mq.OaCOrderPersonal;
import com.xnyzc.lhy.order.entity.order.VehicleType;
import com.xnyzc.lhy.order.entity.param.order.OrderParam;
import com.xnyzc.lhy.order.entity.system.OaSysDispatchLog;
import com.xnyzc.lhy.order.entity.user.OaDUserInfo;
import com.xnyzc.lhy.order.feign.amap.IAmapService;
import com.xnyzc.lhy.order.feign.didi.IDidiService;
import com.xnyzc.lhy.order.mapper.order.OaCOrderPersonalMapper;
import com.xnyzc.lhy.order.mapper.system.OaSysOrderDispatchRecordMapper;
import com.xnyzc.lhy.order.mapper.system.OaSysOrderDriverDispatchHistoryMapper;
import com.xnyzc.lhy.order.mapper.system.OaSysSchedulingStrategyMapper;
import com.xnyzc.lhy.order.mapper.user.OaDUserInfoMapper;
import com.xnyzc.lhy.order.mq.consumer.MqGenerateOrderListConsumer;
import com.xnyzc.lhy.order.service.order.IOaCOrderPersonalDiDiService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: huxuekuo
 * @Date: 2019-07-20 19:29
 * @Description:
 */
@Slf4j
@Component
public class MqOrderConsumer {

    private DefaultMQPushConsumer consumer;

    /**
     * rocketMq 配置文件
     */
    @Autowired
    RocketConfig config;

    /**
     * redis方法封装类
     */
    @Autowired
    RedisCommon redisCommon;

    /**
     * 订单方法封装类
     */
    @Autowired
    OrderCommon orderCommon;

    /**
     * 判断是否开始Mq true 开始 false 关闭
     */
    @Value("${rocketMq.isStart}")
    private boolean isStart;

    /**
     * 司机信息表
     */
    @Autowired
    OaDUserInfoMapper userInfoMapper;

    /**
     * 远程调用资源服务
     */
    @Autowired
    IAmapService amapService;

    /**
     * 远程调用滴滴服务
     */
    @Autowired
    IDidiService didiService;

    /**
     * 滴滴订单service
     */
    @Autowired
    IOaCOrderPersonalDiDiService personalDiDiService;

    /**
     * 调度规则表
     */
    @Autowired
    OaSysSchedulingStrategyMapper oaSysSchedulingStrategyMapper;

    /**
     * 订单表
     */
    @Autowired
    OaCOrderPersonalMapper oaCOrderPersonalMapper;

    /**
     * 调度司机消费者
     */
    @Autowired
    MqGenerateOrderListConsumer orderListConsumer;
    /**
     * 订单调度过程记录
     */
    @Autowired
    OaSysOrderDispatchRecordMapper oaSysOrderDispatchRecordMapper;

    /**
     * 订单调度司机记录表
     */
    @Autowired
    OaSysOrderDriverDispatchHistoryMapper oaSysOrderDriverDispatchHistoryMapper;

    @Autowired
    MqOrderProducer orderProducer;

    @Autowired
    MqDispatchLogProducer dispatchLogProducer;

    private ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 60, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<>());

    /**
     * 开启监听
     */
    @PostConstruct
    public void init() {
        try {
            //判断是否启动RocketMQ
            if (!isStart) {
                return;
            }
            consumer = new DefaultMQPushConsumer(MqConsumerGroupConstant.DISPATCH_ORDER);
            consumer.setNamesrvAddr(config.getNameServerAddr());
            consumer.subscribe(config.getTopic(), MqConstant.ORDER_DISPATCH);
            //批量消费的最大消息条数
            consumer.setConsumeMessageBatchMaxSize(10);
            //设置消费策略:队列头消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            /**
             *  设置监听
             */
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    log.info("当前消费消息--->个数{}", msgs.size());
                    for (MessageExt msg : msgs) {
                        MqOrderConsumer.this.dealBody(msg);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            log.info("---------MQCousumer启动完成------------");
        } catch (Exception e) {
            log.info("---------MQCousumer初始化失败------------");
        }
    }

    /**
     * 处理业务逻辑
     *
     * @param messageExt
     * @return
     */
    public ConsumeConcurrentlyStatus dealBody(MessageExt messageExt) {
        OrderParam orderParam = null;
        try {
            //发送的消息
            String body = new String(messageExt.getBody());
            log.info("给mq发送的消息为：" + body);
            if (CheckUtil.strIsEmpty(body)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            orderParam = JSON.parseObject(body, OrderParam.class);
            //分配车辆
            log.info("开始寻找车辆");
            try {
                OaSysDispatchLog dispatchLog = new OaSysDispatchLog();
                dispatchLog.setOrderId(orderParam.getOrderNo());
                dispatchLog.setTimestamp(System.currentTimeMillis());
                dispatchLog.setCityId(orderParam.getCityId());
                dispatchLog.setLogId(IDUtil.nextId());
                dispatchLogProducer.sendMsg(dispatchLog, EDispatchType.dispatchDriverBefore.getDesc());
                log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,开始调度", orderParam.getOrderNo());
            } catch (Exception e) {
                log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,开始调度,记录订单调度日志失败!", orderParam.getOrderNo());
            }
            findNearCar(orderParam);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    public void findNearCar(OrderParam pOrder) {
        // 司机调度(默认预估调度)
        Integer driverSource = 0;

        int cnt = TimeLimitConstant.FIND_CAR_TIME;
        log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,进入了派单方法", pOrder.getOrderNo());
        while (cnt-- > 0) {
            List<OaDVehicleGps> oaDVehicleGps = orderListConsumer.findVehicle(pOrder);
            log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq, 在redis中获取到了司机列表 司机个数为:[{}]", pOrder.getOrderNo());
            driverSource = 1;
            //如果依然后去到司机列表为空退出本次循环
            if (CheckUtil.collectionIsEmpty(oaDVehicleGps)) {
                log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,获取到符合条件的司机依然为空,退出本次循环!", pOrder.getOrderNo());
                try {
                    Thread.sleep(300);
                } catch (InterruptedException e) {
                    log.error("ORDER_HISTORY 订单号:[{}],调度司机Mq,获取到符合条件的司机依然为空,退出本次循环!", pOrder.getOrderNo());
                    Thread.currentThread().interrupt();
                }
                continue;
            }

            for (OaDVehicleGps gps : oaDVehicleGps) {
                log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,获取司机信息,再次确认是否未接单(Start)!", pOrder.getOrderNo());
                OaDUserInfo oaDUserInfo = userInfoMapper.selectOne(Qw.create().eq(OaDUserInfo.DRIVER_ID, gps.getDriverId()).eq(OaDUserInfo.IS_ORDER, 0));
                if (CheckUtil.objIsEmpty(oaDUserInfo)) {
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,获取司机信息,当前司机已接单循环下一两车(End)", pOrder.getOrderNo());
                    continue;
                }
                log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,获取当前司机ID:[{}],司机接单状态:[{}],司机当前订单号:[{}],详细内容:[{}]",
                        pOrder.getOrderNo(), oaDUserInfo.getDriverId(), oaDUserInfo.getIsOrder(), oaDUserInfo.getCurrentOrderNo(), oaDUserInfo);
                // 查找司机信息
                NettyUserOrderInfo vehicleResult = null;
                // 设置订单信息，用户信息，给司机获取
                NettyDriverOrderInfo userResult = null;
                try {
                    //获取司机信息
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,开始组装司机需要的派单信息(Start)", pOrder.getOrderNo());
                    vehicleResult = orderCommon.vehicleResult(gps, oaDUserInfo);
                    vehicleResult.setStartLat(pOrder.getFlat());
                    vehicleResult.setStartLng(pOrder.getFlng());
                    vehicleResult.setOrderNo(pOrder.getOrderNo());
                    vehicleResult.setOrderStatus(EOrderStatus.driverOrder.getType());
                    // 获取用户信息（需要修改）
                    userResult = userResult(pOrder, gps);
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,组装司机需要的派单信息成功结束(End) ", pOrder.getOrderNo());
                } catch (Exception e) {
                    StackTraceElement stackTrace = e.getStackTrace()[0];
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,组装司机信息错误,错误信息:[{}],错误行数:[{}],错误方法:[{}] ", pOrder.getOrderNo(), e, stackTrace.getLineNumber(), stackTrace.getMethodName());
                    continue;
                }
                VehicleType type = pOrder.getTypeList().get(0);
                //匹配成功
                Integer number = type.getNumber();
                type.setNumber(number--);
                // 找到司机后修改司机信息，锁定司机状态，调用司机抢单接口，等待滴滴回应抢单结果
                log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,开始设置司机司机抢单接口需要的数据(Start)", pOrder.getOrderNo());
                StriveCallbackParam callbackParam = new StriveCallbackParam();
                callbackParam.setOpen_oid(pOrder.getOpenOid());
                callbackParam.setOrder_id(pOrder.getOrderNo());
                callbackParam.setProduct_type(pOrder.getProductType());
                callbackParam.setRide_type(pOrder.getRideType());
                // 设置司机ID
                callbackParam.setDriver_id(String.valueOf(gps.getDriverId()));
                callbackParam.setDriver_phone(vehicleResult.getVehUserPhone());
                callbackParam.setDriver_name(vehicleResult.getVehUserName());
                callbackParam.setDriver_lat(new BigDecimal(gps.getLat()));
                callbackParam.setDriver_lng(new BigDecimal(gps.getLng()));
                callbackParam.setCar_brand(vehicleResult.getBrand());
                callbackParam.setCar_plate_number(vehicleResult.getVehNo());
                callbackParam.setVehicleId(vehicleResult.getVehId());
                // 车辆颜色
                callbackParam.setCar_color(vehicleResult.getCarColor());
                // 司机接单数量
                callbackParam.setFinish_order(vehicleResult.getOrderQuantityCompleted());
                log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,开始设置司机司机抢单接口需要的数据(END),调用抢单接口的数据为:[{}] ", pOrder.getOrderNo(), callbackParam);
                //生成订单
                oaDUserInfo.setIsOrder(1);
                oaDUserInfo.setCurrentOrderNo(vehicleResult.getOrderNo());
                if (userInfoMapper.updateById(oaDUserInfo) > 0) {
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,锁定司机成功,开始进行更改订单信息(Start)  ", pOrder.getOrderNo());
                    // ##################################### 给订单表添加司机信息
                    try {
                        QueryWrapper<OaCOrderPersonal> queryWrapper = new QueryWrapper<>();
                        queryWrapper.select(OaCOrderPersonal.VERSION);
                        queryWrapper.eq(OaCOrderPersonal.ORDER_NO, vehicleResult.getOrderNo());
                        OaCOrderPersonal one = personalDiDiService.getOne(queryWrapper);
                        // 用version加锁
                        UpdateWrapper<OaCOrderPersonal> updateWrapper = new UpdateWrapper<>();
                        updateWrapper.set(OaCOrderPersonal.DRIVER_ID, vehicleResult.getDriverId());
                        updateWrapper.set(OaCOrderPersonal.VEHICLE_ID, vehicleResult.getVehId());
                        updateWrapper.set(OaCOrderPersonal.VERSION, one.getVersion());
                        updateWrapper.set(OaCOrderPersonal.UPDATE_TIME, new Date());
                        updateWrapper.eq(OaCOrderPersonal.ORDER_NO, vehicleResult.getOrderNo());
                        personalDiDiService.update(updateWrapper);
                        log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,锁定司机成功,更改订单信息成功(END)  ", pOrder.getOrderNo());
                    } catch (Exception e) {
                        throw DiDiException.create(EDiDiErrorCode.grabSheetAddMsgFail);
                    }
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,向redis中保存司机需要的信息(Start)  ", pOrder.getOrderNo());
                    userResult.setOrderStatus(4);
                    redisCommon.set(String.format("%s%s", RedisPrefixConstant.ORDER_VEH_TEMPORARY_STORAGE, callbackParam.getOrder_id()), JSON.toJSONString(userResult), 86400);
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,向redis中保存司机需要的信息(End)  ", pOrder.getOrderNo());
                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,开始调用抢单接口(Start)  ", pOrder.getOrderNo());
                    callbackParam.setCityId(pOrder.getCityId());
                    try {
                        OaSysDispatchLog dispatchLog = new OaSysDispatchLog();
                        dispatchLog.setType(EDispatchType.dispatchDriverAfter.getType());
                        dispatchLog.setOrderId(pOrder.getOrderNo());
                        dispatchLog.setDriverId(gps.getDriverId());
                        dispatchLog.setDriverLat(Double.valueOf(gps.getLat()));
                        dispatchLog.setDriverLng(Double.valueOf(gps.getLng()));
                        dispatchLog.setDispatchDistance(Double.valueOf(userResult.getGouldDistanceKm()));
                        dispatchLog.setTimestamp(System.currentTimeMillis());
                        dispatchLog.setCityId(pOrder.getCityId());
                        dispatchLog.setLogId(IDUtil.nextId());
                        dispatchLog.setDriverSource(driverSource);
                        dispatchLogProducer.sendMsg(dispatchLog, EDispatchType.dispatchDriverAfter.getDesc());
                        log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,找到司机调度结束", pOrder.getOrderNo());
                    } catch (Exception e) {
                        log.error("记录订单调度日志!");
                    }
                    //司机抢单
                    didiService.getStriveCallback(callbackParam);

                    log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,调用抢单接口结束,请求参数:[{}]", pOrder.getOrderNo(), callbackParam);
                    return;
                }
            }
        }
        try {
            OaSysDispatchLog dispatchLog = new OaSysDispatchLog();
            dispatchLog.setOrderId(pOrder.getOrderNo());
            dispatchLog.setTimestamp(System.currentTimeMillis());
            dispatchLog.setCityId(pOrder.getCityId());
            dispatchLog.setLogId(IDUtil.nextId());
            dispatchLogProducer.sendMsg(dispatchLog, EDispatchType.dispatchDriverTimeOut.getDesc());
            log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,调度超时", pOrder.getOrderNo());
        } catch (Exception e) {
            log.info("ORDER_HISTORY 订单号:[{}],调度司机Mq,记录订单调度日志失败", pOrder.getOrderNo());
        }
    }

    /**
     * 封装乘客信息(需要修改)
     *
     * @return
     */
    public NettyDriverOrderInfo userResult(OrderParam pOrder, OaDVehicleGps gps) {
        NettyDriverOrderInfo driverOrderInfo = orderCommon.getUserMessage(pOrder, gps);
        driverOrderInfo.setStartLat(pOrder.getFlat());
        driverOrderInfo.setStartLng(pOrder.getFlng());
        driverOrderInfo.setEndLat(pOrder.getTlat());
        driverOrderInfo.setEndLng(pOrder.getTlng());
        return driverOrderInfo;
    }

    //
    public void dispatchOrder() {
//        Set<String> keys = redisCommon.keys(RedisPrefixConstant.ORDER_DISPATCH_INFO + "*");
//        for (String key : keys) {
//            HashMap<String, String> hashMap = redisCommon.getHashMap(key);
//            // 获取调度时间戳
//            String time = hashMap.get("time");
//            Long timestart = Long.valueOf(time);
//            // 时间差
//            long minute = System.currentTimeMillis() - timestart;
//            // 时间差大于等于 5 分钟删除redis key,停止循环
//            if (minute >= (5 * 60 * 1000)) {
//                redisCommon.remove(key);
//                continue;
//            }
//            // 未超时,调度找车
//            OrderParam orderParam = JSON.parseObject(hashMap.get("orderJson"), OrderParam.class);
//            executor.submit(new Runnable() {
//                @Override
//                public void run() {
//                    try {
//                        findNearCar(orderParam);
//                    } catch (Exception e) {
//                    }
//                }
//            });
//
    }
//
//    }

}










